import pandas as pd
from hdfs.client import Client
import pymysql

#   修改pandas的内置参数，让其显示所有的行和列
#   显示所有列数据
pd.set_option('display.max_columns',None)
#   显示所有行数据
pd.set_option('display.max_rows',None)
#   设置value的长度
pd.set_option('max_colwidth',100)
#   设置1000列时才换行
pd.set_option('display.width',1000)

#   设置读取文件时的列名
cols=['id','title','house_trait','money','landlord','plot','address','other_trait','url']

#   names指定列名
data=pd.read_csv('data01.csv',names=cols)
#   axis=1指定竖着查询，是否含有id列，有的话就删除
data.drop('id',axis=1,inplace=True)
# print(data)
#   去除重复的数据
data.drop_duplicates(inplace=True)

#   1.将每个月多少租金一列的单位去掉(因为都是按多少钱一个月)
data['money']=data['money'].apply(lambda x:str(x).split('元/')[0])
# print(data['money'])


#   2.  将房屋的型号,面积,楼层单独成列
#   房型(几室几厅):house_type
data['house_type']=data['house_trait'].apply(
    lambda x:
    (str(x).replace('[','').replace(']','').split(',')[0]+str(x).replace('[','').replace(']','').split(',')[1]).replace("'",'')
    .replace(' ','')
)
# print(data['house_type'])

#   面积:area
data['area']=data['house_trait'].apply(lambda x:str(x).replace('[','').replace(']','').split(',')[2].replace("'",'').replace('平米',''))
# print(data['area'])

#   楼层floor
data['floor']=data['house_trait'].apply(lambda x:str(x).replace('[','').replace(']','').split(',')[3].replace("'",''))
# print(data['floor'])

#   将市区字段单独成列(downtown:市区)
data['downtown']=data['address'].apply(lambda x:str(x).replace('[','').replace(']','').split(',')[0].replace("'",''))
# print(data['downtown'])

#   将出租类型单独成列(rent_out_type:出租类型)
data['rent_out_type']=data['other_trait'].apply(lambda x:str(x).replace('[','').replace(']','').split(',')[0].replace("'",''))
# print(data['rent_out_type'])

#   把南北朝向单独成列(orientation:朝向)
data['orientation']=data['other_trait'].apply(lambda x:str(x).replace('[','').replace(']','').split(',')[1].replace("'",''))
# print(data['orientation'])

#   将有电梯的数据单独成列(elevator:电梯)(有的为1，没有的为0)
data['elevator']=data['other_trait'].apply(
    lambda x:
    1 if "'有电梯'" in str(x) else 0
)
# print(data['elevator'])

#   将是否有地铁单独成列(subway:地铁)
#   定义处理数据的函数
def get_subway(x):
    for j in  str(x).replace('[','').replace(']','').split(','):
        if '线' in j:
            return j
    else:
        return 0
data['subway']=data['other_trait'].apply(get_subway)
print(len(data))

#   最后将处理完的数据保存成文件
data.to_csv('data02.csv',index=False)
print("处理完的数据已经保存到本地,文件名为data02.csv................")

#  并且将处理完的数据上传到hdfs
#   创建一个hdfs的客户端实例(9870是web访问hdfs的端口号)
client=Client('http://192.168.40.110:9870')
#   hdfs上面存储的路径
hdfs_path="/qcl/"
#   使用客户端的upload方法将文件上传到hdfs
client.upload(hdfs_path=hdfs_path,local_path='data02.csv',overwrite=True)
print("处理完的数据已经保存到hdfs,保存的路径为:/qcl/data02.csv........................")

#     将处理好的数据写入mysql
#   建立与mysql的连接
connect=pymysql.connect(host='192.168.40.110',user='root',password='123456',database='qcl',charset='utf8')
#   通过连接创建游标对象
cursor=connect.cursor()
#   定义sql如果存在表就删除表(data_processing:数据处理)
drop_table_sql='drop table if exists data_processing;'
cursor.execute(drop_table_sql)
#   定义创建表格的sql
CreateTable_sql="""
create table if not exists data_processing(
            id int primary key AUTO_INCREMENT,
            title varchar(500),
            house_trait varchar(500),
            money int,
            landlord varchar(500),
            plot varchar(500),
            address varchar(500),
            other_trait varchar(500),
            url varchar(500),
            house_type varchar(500),
            area int,
            floor varchar(500),
            downtown varchar(500),
            rent_out_type varchar(500),
            orientation varchar(500),
            elevator int,
            subway varchar(500)
);
"""
#   执行创建表的语句
cursor.execute(CreateTable_sql)
#   定义插入数据的sql语句
insert_sql="""
    insert into data_processing(
         title,house_trait,money,landlord,plot,address,other_trait,url,house_type,area,floor,downtown,rent_out_type,
         orientation,elevator,subway
    )
    values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);
"""
#   换成list[row]的形式方便写入mysql
data_list=data.values.tolist()
#   执行插入数据
cursor.executemany(insert_sql,data_list)

#   提交任务(只有提交了任务才会执行sql)
connect.commit()
#   关闭游标对象
cursor.close()
#   关闭连接
connect.close()
print("处理完的数据已经保存到mysql的qcl数据库的data_processing表..............")



